package com.blyat.xsoft.kernel.component.mq.rocketmq.consumer.push;

import com.alibaba.fastjson.JSON;
import com.blyat.xsoft.kernel.component.mq.rocketmq.Carrier;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.message.MessageExt;

import java.util.List;
import java.util.stream.Collectors;

/**
 * @Auther: syh
 * @Date: 2020/8/7
 * @Description:
 */
public abstract class ConcurrentlyPushConsumer<T> extends PushConsumer<T> implements MessageListenerConcurrently {
    public ConcurrentlyPushConsumer(String topic) {
        super(topic);
    }

    public ConcurrentlyPushConsumer(String topic, int batchSize) {
        super(topic, batchSize);
    }

    public ConcurrentlyPushConsumer(String groupName, String topic, int batchSize) {
        super(groupName, topic, batchSize);
    }

    public ConcurrentlyPushConsumer(String groupName, String topic, String tags, int batchSize) {
        super(groupName, topic, tags, batchSize);
    }

    @Override
    public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> messageList, ConsumeConcurrentlyContext context) {
        List<T> list = messageList.parallelStream()
                .map(message -> {
                    Carrier<T> carrier = JSON.parseObject(new String(message.getBody()), Carrier.class);
                    return carrier.getContent();
                }).collect(Collectors.toList());

        try {
            consume(list);
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        } catch (Exception e) {
            return ConsumeConcurrentlyStatus.RECONSUME_LATER;
        }
    }
}
